// Copyright (c) 2021 Presto Labs Pte. Ltd.
// Author: xguo

#include "coin2/exchange/base/order/parser.h"

#include "coin2/exchange/base/order/order_context_manager.h"
#include "coin2/exchange/base/order/utils.h"
#include "quickfix/Message.h"

namespace coin2::exchange::base::order {

bool BaseParser::IsValidResponse(const std::shared_ptr<AsioHttpContext>& http_context) {
  using Status = AsioHttpContextStatus;
  if (!http_context) {
    return false;
  }
  const auto& status_code = http_context->res.result_int();
  if (status_code >= 500) {
    return false;
  }
  switch (http_context->status) {
    case Status::kResponded:
      return true;
      break;
    case Status::kError:
      return status_code != 200;
      break;
    case Status::kTimeout:
      return status_code != 200 || !http_context->res.body().empty();
      break;
    default:
      return false;
      break;
  }
}

bool BaseParser::PrepareJsonDocument(JsonDocument* document) {
  if (msg_ == nullptr) {
    return false;
  }
  switch (sub_topic_id_) {
    case OrderSubTopicId::QUICKFIX_MESSAGE:
      // pass
      break;
    case OrderSubTopicId::WS_MESSAGE:
      document->Parse<rapidjson::kParseNumbersAsStringsFlag>(msg_->data->data());
      break;
    default:
      document->Parse<rapidjson::kParseNumbersAsStringsFlag>(msg_->http_context->res.body().data());
      break;
  }
  return !document->HasParseError();
}

void BaseParser::UpdateTimestamp(int64_t ts) {
  CHECK_GT(ts, 0);
  if (ts >= current_ts_) {
    current_ts_ = ts;
  } else {
    LOG(WARNING) << "time reversed! " << current_ts_ << ", " << ts;
  }
}

int64_t BaseParser::GetCurrentTimestamp() const {
  CHECK_GT(current_ts_, 0);
  return current_ts_;
}

std::string BaseParser::RawMessage::DebugString() const {
  auto raw_msg = ToProto();
  return raw_msg.DebugString();
}

std::string BaseParser::RawMessage::SerializeToString() const {
  std::string bytes;
  auto raw_msg = ToProto();
  CHECK(raw_msg.SerializeToString(&bytes));
  return bytes;
}

auto BaseParser::RawMessage::ToProto() const -> RawMessageProto {
  const auto sub_topic_id = static_cast<OrderSubTopicId>(type);
  RawMessageProto raw_msg;
  std::string s;

  raw_msg.set_type(type);
  raw_msg.set_raw_msg_id(raw_msg_id);
  raw_msg.set_receive_timestamp(timestamp);
  raw_msg.set_mea(mea.String());

  // parsed result
  if (type >= 80 && type <= 91) {
    CHECK(proto_msg->SerializeToString(&s));
    raw_msg.set_raw_bytes(s);
    return raw_msg;
  }

  // REST_MESSAGE
  if (EqualToAnyOf(
          sub_topic_id,
          OrderSubTopicId::REST_MESSAGE,
          OrderSubTopicId::LISTEN_KEY,
          OrderSubTopicId::REST_ACCOUNTS,
          OrderSubTopicId::REST_ACCOUNT_BALANCE,
          OrderSubTopicId::REST_ACCOUNT_POSITION,
          OrderSubTopicId::REST_ALL_ORDERS,
          OrderSubTopicId::REST_OPEN_ORDERS,
          OrderSubTopicId::REST_CANCEL_ORDERS,
          OrderSubTopicId::REST_FILL_ORDERS,
          OrderSubTopicId::REST_DONE_ORDERS,
          OrderSubTopicId::REST_ORDERS_INFO,
          OrderSubTopicId::SUBMIT_RESPONSE,
          OrderSubTopicId::CANCEL_RESPONSE,
          OrderSubTopicId::AMEND_RESPONSE,
          OrderSubTopicId::MULTI_CANCEL_RESPONSE)) {
    OrderExecutionProto order_execution;
    *raw_msg.mutable_http_context() = http_context->ToProto();

    if (oc) {
      *order_execution.mutable_order_context() = oc->ToProto();
      CHECK(order_execution.SerializeToString(&s));
    }

    raw_msg.set_raw_bytes(s);
    *raw_msg.mutable_http_context() = http_context->ToProto();

    for (const auto* tmp_oc : oc_list) {
      auto* proto_oc = order_execution.add_order_contexts();
      *proto_oc = tmp_oc->ToProto();
    }

    CHECK(order_execution.SerializeToString(&s));
    raw_msg.set_raw_bytes(s);
    *raw_msg.mutable_http_context() = http_context->ToProto();
    return raw_msg;
  }

  // SYSTEM
  if (sub_topic_id == OrderSubTopicId::SYSTEM) {
    *raw_msg.mutable_error_code() = error_code;
    return raw_msg;
  }

  // ORDER_EXECUTION
  if (sub_topic_id == OrderSubTopicId::ORDER_EXECUTION) {
    *raw_msg.mutable_order_execution() = RawMessage::order_execution;
    return raw_msg;
  }

  // WS_MESSAGE
  if (sub_topic_id == OrderSubTopicId::WS_MESSAGE) {
    raw_msg.set_raw_bytes(*data);
    return raw_msg;
  }

  // QUICKFIX MESSAGE
  if (sub_topic_id == OrderSubTopicId::QUICKFIX_MESSAGE) {
    raw_msg.set_raw_bytes(fix_message->toString());
    return raw_msg;
  }

  // GRPC MESSAGE
  if (sub_topic_id == OrderSubTopicId::GRPC_MESSAGE) {
    raw_msg.set_raw_bytes(grpc_context->ToJsonString());
    return raw_msg;
  }

  // RAW MESSAGE
  if (sub_topic_id == OrderSubTopicId::RAW_BUFFER) {
    raw_msg.set_raw_bytes(raw_buffer.data(), raw_buffer.size());
    return raw_msg;
  }

  CHECK(false) << "unknown sub_topic_id: " << sub_topic_id;
  return raw_msg;
}

BaseParser::RawMessage BaseParser::RawMessage::FromProto(const RawMessageProto& proto_msg) {
  using OrderContext = coin2::exchange::base::order::OrderContext;
  CHECK(proto_msg.has_mea()) << proto_msg.DebugString();

  RawMessage msg;
  msg.type = proto_msg.type();
  msg.timestamp = proto_msg.receive_timestamp();  // must set timestamp.
  msg.raw_msg_id = proto_msg.raw_msg_id();
  msg.mea = MarketExchangeApi::FromString(proto_msg.mea());

  switch (proto_msg.type()) {
    // Handle rest events
    case OrderSubTopicId::REST_MESSAGE:
    case OrderSubTopicId::LISTEN_KEY:
    case OrderSubTopicId::REST_ACCOUNTS:
    case OrderSubTopicId::REST_ACCOUNT_BALANCE:
    case OrderSubTopicId::REST_ACCOUNT_POSITION:
    case OrderSubTopicId::REST_ALL_ORDERS:
    case OrderSubTopicId::REST_OPEN_ORDERS:
    case OrderSubTopicId::REST_CANCEL_ORDERS:
    case OrderSubTopicId::REST_FILL_ORDERS:
    case OrderSubTopicId::SUBMIT_RESPONSE:
    case OrderSubTopicId::AMEND_RESPONSE:
    case OrderSubTopicId::CANCEL_RESPONSE:
    case OrderSubTopicId::MULTI_CANCEL_RESPONSE: {
      CHECK(!proto_msg.has_http_response());
      msg.http_context = AsioHttpContext::FromProto(proto_msg.http_context());

      OrderExecutionProto execution;
      if (proto_msg.has_raw_bytes()) {
        execution.ParseFromString(proto_msg.raw_bytes());
        if (execution.has_order_context()) {
          msg.oc_list_tmp.emplace_back(OrderContext::FromProto(execution.order_context()));
          msg.oc = msg.oc_list_tmp.back().get();
          msg.oc_list.push_back(msg.oc);
        } else {
          CHECK(execution.order_contexts_size() > 0) << execution.DebugString();
          for (const auto& proto_oc : execution.order_contexts()) {
            msg.oc_list_tmp.emplace_back(OrderContext::FromProto(proto_oc));
            msg.oc_list.push_back(msg.oc_list_tmp.back().get());
            msg.http_context = AsioHttpContext::FromProto(proto_msg.http_context());
          }
        }
      }
      break;
    }
    // Handle ws events
    case OrderSubTopicId::WS_MESSAGE: {
      msg.data_tmp = proto_msg.raw_bytes();
      msg.data = &msg.data_tmp;
      break;
    }
    case OrderSubTopicId::SYSTEM: {
      msg.error_code = proto_msg.error_code();
      break;
    }
    case OrderSubTopicId::ORDER_EXECUTION: {
      msg.order_execution = proto_msg.order_execution();
      break;
    }
    case OrderSubTopicId::QUICKFIX_MESSAGE: {
      msg.fix_message_ptr.reset(new FIX::Message(proto_msg.raw_bytes()));
      msg.fix_message = msg.fix_message_ptr.get();
      break;
    }
    default: {
      if (!(msg.type >= 80 && msg.type < 90)) {
        CHECK(false) << proto_msg.DebugString();
      }
    }
  }
  return msg;
}

}  // namespace coin2::exchange::base::order
